package com.hyts.rxjava.sample;

import rx.Observable;

import java.util.concurrent.TimeUnit;

/**
 * project-name:coffice-toolbox
 * package-name:com.hyts.rxjava.list
 * author:Libo/Alex
 * create-date:2019-06-27 11:10
 * email:liboware@gmail.com
 * description:此类主要用于:
 */
public class Streams {


    public static void main(String[] args){
        /*//创建observer对象消息发布者
        Observable.create(param->{
            param.onStart();
            param.onNext("测试");
            param.onCompleted();
        }).subscribe(System.out::println);
        //测试输出操作
        Observable.just("测试").subscribeOn(Schedulers.immediate()).subscribe(System.out::println);
        //测试输出操作
        Observable.create(subscriber -> {
            new Thread(() -> {
                try {
                    subscriber.onNext("测试");
                    subscriber.onCompleted();
                } catch (Exception e) {
                    subscriber.onError(e);
                }
            }).start();
        }).subscribe(System.out::println);
        // add error handling
        Observable.create(subscriber -> {
            try {
                subscriber.onNext(1/0);
                subscriber.onError(new NullPointerException());
                subscriber.onCompleted();
            } catch (Exception e) {
                subscriber.onError(e);
            }
        }).subscribeOn(Schedulers.io())
                .map(data -> data + " --> at " + System.currentTimeMillis())
                .onErrorResumeNext(Observable.just("1231"))
                .subscribe(System.out::println);*/

        // infinite
       /* Observable.create(subscriber -> {
            int i = 0;
            while (!subscriber.isUnsubscribed()) {
                subscriber.onNext(i++);
            }
        }).take(10).subscribe(System.out::println);*/

        //Hello World
        /*Observable observable  = Observable.just("fallback value");

        Observable.create(subscriber -> {
            throw new RuntimeException("failed!");
        }).onErrorResumeNext(observable).subscribe(System.out::println);*/

        /*Observable.create(subscriber -> {
            throw new RuntimeException("failed!");
        }).onErrorReturn(throwable -> {
            return "fallback value";
        }).onErrorResumeNext(Observable.just("123")).subscribe(System.out::println);
*/

       /* Observable.create(subscriber -> {
            throw new RuntimeException("failed!");
        }).retryWhen(attempts -> {
            return attempts.zipWith(Observable.range(1, 3), (throwable, i) -> i)
                    .flatMap(i -> {
                        if(i % 2!=0){
                            throw new RuntimeException();
                        }
                        System.out.println("delay retry by " + i + " second(s)");
                        return Observable.timer(i, TimeUnit.SECONDS);
                    }).concatWith(Observable.error(new RuntimeException("Exceeded 3 retries")));
        }).subscribe(System.out::println, t -> t.printStackTrace());*/

    }

}
